/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package aiaudio.processing.reduce;

import aiaudio.processing.reduce.klsh.kmeaning.DictionaryCoordinatePair;
import aiaudio.processing.reduce.klsh.kmeaning.mappers.KMeansAlgorithmMapper;
import aiaudio.processing.reduce.klsh.kmeaning.mappers.KMeansDictionaryMapper;
import aiaudio.processing.reduce.klsh.kmeaning.reducers.KMeansAlgorithmReducer;
import aiaudio.processing.reduce.klsh.kmeaning.reducers.KMeansDictionaryReducer;
import aiaudio.lastfm.hbase.CannotCreateTableException;
import aiaudio.Application;
import aiaudio.processing.MainTableGroup;
import aiaudio.processing.base.TimeMeasuredAlgorithm;
import java.io.IOException;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.mapreduce.Job;

/**
 *
 * @author nastya
 */
public class KMeansDimensionReducingAlgorithm extends TimeMeasuredAlgorithm<KMeansDimensionReducingParameters> {

    public KMeansDimensionReducingAlgorithm(MainTableGroup mainTableGroup) {
        super(mainTableGroup, "KMeansDimensionReducingAlgorithm");
    }

    @Override
    protected void createTemporaryTables() throws IOException {
    }

    @Override
    protected void measureAll() throws IOException {
    }

    @Override
    protected void process() throws IOException, CannotCreateTableException, InterruptedException, ClassNotFoundException {
        Job job = collectDictionariesJob();
        job.waitForCompletion(true);
        job = calculateResultJob();
        job.waitForCompletion(true);

    }

    @Override
    protected void removeTemporaryTables() {
    }

    @Override
    protected void writeParameters(KMeansDimensionReducingParameters parameters) throws IOException {
    }

    private Job collectDictionariesJob() throws IOException {
        Job job = new Job();

        job.getConfiguration().setLong(KMeansDictionaryMapper.PARAM_TEST_SET_SIZE, parameters.getTestSetSize());
        job.getConfiguration().setLong(KMeansDictionaryMapper.PARAM_DICTIONARY_SIZE, parameters.getDictionarySize());
        job.getConfiguration().setLong(KMeansDictionaryMapper.PARAM_DICTIONARY_COUNT, parameters.getDictionaryCount());

        Scan scan = new Scan();

        String tracksTableName = Application.database().getByName(mainTableGroup.getInitialDataTableGroup().getArtistsTable()).getDatabaseName();
        TableMapReduceUtil.initTableMapperJob(tracksTableName, scan, KMeansDictionaryMapper.class, ImmutableBytesWritable.class, ImmutableBytesWritable.class, job);

        String resTableName = Application.database().getByName(mainTableGroup.getReduceArtistsAlgTableGroup().getDictionaryTable()).getDatabaseName();
        TableMapReduceUtil.initTableReducerJob(resTableName, KMeansDictionaryReducer.class, job);
        return job;
    }

    private Job calculateResultJob() throws IOException {
        Job job = new Job();

        job.getConfiguration().set(KMeansAlgorithmMapper.DICTIONARY_TABLE_NAME, mainTableGroup.getReduceArtistsAlgTableGroup().getDictionaryTable());
        job.getConfiguration().set(KMeansAlgorithmMapper.RATINGS_SOURCE_NAME, mainTableGroup.getSplitDataSetAlgTableGroup().getTrainingRatingMatrix());

        Scan scan = new Scan();

        String ratingsTableName = Application.database().getByName(mainTableGroup.getSplitDataSetAlgTableGroup().getTrainingRatingMatrix()).getDatabaseName();
        TableMapReduceUtil.initTableMapperJob(ratingsTableName, scan, KMeansAlgorithmMapper.class, ImmutableBytesWritable.class, DictionaryCoordinatePair.class, job);

        String resTableName = Application.database().getByName(mainTableGroup.getReduceArtistsAlgTableGroup().getReducedTable()).getDatabaseName();
        TableMapReduceUtil.initTableReducerJob(resTableName, KMeansAlgorithmReducer.class, job);

        return job;
    }
}
